本申请提供了一种并行处理消息的方法、装置、节点及服务器集群,所述方法包括:从预置的配置数据库中获取线程分配规则和消息分配规则;所述线程分配规则用于表示集群中的每个处理节点所拥有的线程个数,所述消息分配规则用于表示待处理的消息由哪个线程进行处理;依据所述线程分配规则创建每个处理节点对应的多个线程;触发所述创建的多个线程按照所述消息分配规则从消息源领取消息,并进行处理。采用本申请实施例提供的方法、装置、节点或服务器集群,可以解决现有技术中单线程领取消息导致的影响消息处理的实时性和服务器的吞吐量的技术问题。
1.一种并行处理消息的方法,其特征在于,该方法包括:从预置的配置数据库中获取线程分配规则和消息分配规则;所述线程分配规则用于表示集群中的每个处理节点所拥有的线程个数,所述消息分配规则用于表示待处理的消息由哪个线程进行处理;依据所述线程分配规则创建每个处理节点对应的多个线程;触发所述创建的多个线程按照所述消息分配规则从消息源领取消息,并进行处理;所述线程按照所述消息分配规则从消息源中领取对应的消息,包括:将请求处理消息的用户标识编号按照所述线程个数进行取模运算;线程编号与所述运算结果匹配的线程从所述消息源中领取所述用户标识触发的消息,其中,同一个所述用户标识触发的消息在同一个顺序子集中。
2.根据权利要求1所述的方法,其特征在于,还包括:对所述线程分配规则和消息分配规则进行更新。
3.根据权利要求1所述的方法,其特征在于,还包括:依据所述处理节点的CPU数量和/或内存参数设置所述线程分配规则。
4.一种并行处理消息的装置,其特征在于,该装置包括:获取模块,用于从预置的配置数据库中获取线程分配规则和消息分配规则;所述线程分配规则用于表示集群中的每个处理节点所拥有的线程个数,所述消息分配规则用于表示待处理的消息由哪个线程进行处理;创建模块,用于依据所述线程分配规则创建每个处理节点对应的多个线程;触发模块,用于触发所述创建的多个线程按照所述消息分配规则从消息源领取消息,并进行处理;所述触发模块包括运算子模块和触发子模块,其中:所述运算子模块用于将请求处理消息的用户标识编号按照所述线程个数进行取模运算;所述触发子模块用于触发线程编号与所述运算结果匹配的线程从所述消息源中领取所述用户标识触发的消息;其中,同一个所述用户标识触发的消息在同一个顺序子集中。
5.根据权利要求4所述的装置,其特征在于,还包括:更新模块,用于对所述线程分配规则和消息分配规则进行更新。
6.根据权利要求4所述的装置,其特征在于,还包括:设置模块,用于依据所述处理节点的CPU数量和/或内存参数设置所述线程分配规则。
7.一种并行处理消息的节点,其特征在于,所述节点包括:权利要求4~6任一项所述装置。
8.一种服务器集群,其特征在于,所述集群包括:至少两个权利要求7所述的并行处理消息的节点。技术领域
本申请涉及网络数据处理领域,特别涉及一种并行处理消息的方法、装置、节点及服务器集群。
背景技术
在P4P(Payxa0forxa0performance,按效果付费)系统可以处理P4P用户在竞价平台(BP)中由于操作推广信息产生的消息,并将该消息实时更新到搜索引擎中,以使用户快速完成信息在搜索引擎中的推广。由于竞价平台是提供给用户使用的并发WEB系统,所以用户消息量相对较大,并且对于单个用户的消息一定要保证时间上的顺序性,所以如何保证整个P4P系统的实时性和吞吐量就成为一个难题。现有技术中,在保证P4P系统的实时性和吞吐量时,一般通过如下方法:预先将消息配置为按线程的不同分为各个不同的组,一组消息对应一个处理该组消息的独立线程;当P4P系统接收到用户发送的处理消息的请求时,每个独立线程都领取预先给其配置的处理消息来执行。其中,一个独立线程对应的消息组中的所有消息处理按时间顺序,从而保证了局部一致性;而多个独立线程之间则完全并行处理。其中,线程是操作系统中CPU调度的基本单位,代表着一个CPU作业过程。而并行的含义则是,让多个处理单元(例如线程)同时独立进行业务计算,并行机制使得现在的计算机应用吞吐量得到大幅提升。上述的局部一致性可以理解为,在整个消息大集合中,存在很多小集合,而这些小集合是按一定业务规则(例如按线程)划分的,各个小集合内部的消息处理顺序需要与消息产生的时间顺序一致,但不同小集合之间的消息处理是没有前后顺序的,这就是局部一致性。但是上述现有技术存在一个问题,就是存在一个专门领取消息的独立线程,该独立线程将消息按顺序子集分组,再对应分给多线程并行执行。这就使得在多个线程处理消息之前的领取过程,会在消息海量增长的时候,变得非常缓慢和耗时,这势必会影响消息处理的实时性和服务器的吞吐量。总之,目前需要本领域技术人员迫切解决的一个技术问题就是:如何能够创新的提出一种并行处理消息的方法,以解决现有技术中单线程领取消息导致的影响消息处理的实时性和服务器的吞吐量的技术问题。
发明内容
本申请所要解决的技术问题是提供一种并行处理消息的方法,用以解决现有技术中单线程领取消息导致的影响消息处理的实时性和服务器的吞吐量的技术问题。本申请还提供了一种并行处理消息的装置、节点及服务器集群,用以保证上述方法在实际中的实现及应用。为了解决上述问题,本申请公开了一种并行处理消息的方法,包括:从预置的配置数据库中获取线程分配规则和消息分配规则;所述线程分配规则用于表示集群中的每个处理节点所拥有的线程个数,所述消息分配规则用于表示待处理的消息由哪个线程进行处理;依据所述线程分配规则创建每个处理节点对应的多个线程;触发所述创建的多个线程按照所述消息分配规则从消息源领取消息,并进行处理。优选的,一个所述线程按照所述消息分配规则从消息源中领取对应的消息,包括:将请求处理消息的用户标识编号按照所述线程个数进行取模运算;线程编号与所述运算结果匹配的线程从所述消息源中领取所述用户标识触发的消息。优选的,还包括:对所述线程分配规则和消息分配规则进行更新。优选的,还包括:依据所述处理节点的CPU数量和/或内存参数设置所述线程分配规则。本申请公开了一种并行处理消息的装置,包括:获取模块,用于从预置的配置数据库中获取线程分配规则和消息分配规则;所述线程分配规则用于表示集群中的每个处理节点所拥有的线程个数,所述消息分配规则用于表示待处理的消息由哪个线程进行处理;创建模块,用于依据所述线程分配规则创建每个处理节点对应的多个线程;触发模块,用于触发所述创建的多个线程按照所述消息分配规则从消息源领取消息,并进行处理。优选的,所述触发模块包括运算子模块和触发子模块,其中:所述运算子模块用于将请求处理消息的用户标识编号按照所述线程个数进行取模运算;所述触发子模块用于触发线程编号与所述运算结果匹配的线程从所述消息源中领取所述用户标识触发的消息。优选的,还包括:更新模块,用于对所述线程分配规则和消息分配规则进行更新。优选的,还包括:设置模块,用于依据所述处理节点的CPU数量和/或内存参数设置所述线程分配规则。本申请公开了一种并行处理消息的节点,包括:前述任一项并行处理消息的装置。本申请公开了一种服务器集群,包括:至少两个前述的并行处理消息的节点。与现有技术相比,本申请包括以下优点:在本申请中,通过建立线程与包括其需要处理消息的顺序子集的关联,从而能够实现线程领取消息过程的并行性,提高了节点处理消息的实时性,从而带来服务器集群的吞吐量和实时性的提升。进一步的,还能够更为方便和高效的实现对服务器集群扩容,更加适应于目前网络应用场景。当然,实施本申请的任一产品并不一定需要同时达到以上所述的所有优点。
附图说明
为了更清楚地说明本申请实施例中的技术方案,下面将对实施例描述中所需要使用的附图作简单地介绍,显而易见地,下面描述中的附图仅仅是本申请的一些实施例,对于本领域普通技术人员来讲,在不付出创造性劳动性的前提下,还可以根据这些附图获得其他的附图。图1是本申请的一种并行处理消息的方法实施例1的流程图;图2是方法实施例1中步骤103的流程图;图3是本申请的一种并行处理消息的方法实施例2的流程图;图4是本申请的方法实施例2在实际应用中的结构框图;图5是本申请的一种并行处理消息的装置实施例1的结构框图;图6是装置实施例1中触发模块503的结构框图;图7是本申请的一种并行处理消息的装置实施例2的结构框图。
具体实施方式
下面将结合本申请实施例中的附图,对本申请实施例中的技术方案进行清楚、完整地描述,显然,所描述的实施例仅仅是本申请一部分实施例,而不是全部的实施例。基于本申请中的实施例,本领域普通技术人员在没有做出创造性劳动前提下所获得的所有其他实施例,都属于本申请保护的范围。本申请可用于众多通用或专用的计算装置环境或配置中。例如:个人计算机、服务器计算机、手持设备或便携式设备、平板型设备、多处理器装置、包括以上任何装置或设备的分布式计算环境等等。本申请可以在由计算机执行的计算机可执行指令的一般上下文中描述,例如程序模块。一般地,程序模块包括执行特定任务或实现特定抽象数据类型的例程、程序、对象、组件、数据结构等等。也可以在分布式计算环境中实践本申请,在这些分布式计算环境中,由通过通信网络而被连接的远程处理设备来执行任务。在分布式计算环境中,程序模块可以位于包括存储设备在内的本地和远程计算机存储介质中。本申请的主要思想之一可以包括,通过建立线程与包括其需要处理消息的顺序子集的关联,从而能够实现线程领取消息过程的并行性,提高了节点处理消息的实时性,从而带来服务器集群的吞吐量和实时性的提升。参考图1,示出了本申请一种并行处理消息的方法实施例1的流程图,可以包括以下步骤:步骤101:从预置的配置数据库中获取线程分配规则和消息分配规则;所述线程分配规则用于表示集群中的每个处理节点所拥有的线程个数,所述消息分配规则用于表示待处理的消息由哪个线程进行处理。本申请实施例中预置的配置数据库,专门用来保存预先配置的线程分配规则和消息分配规则,所述线程分配规则用于表示集群中的每个处理节点所拥有的线程个数,例如,当线程分配规则表示的集群中的每个处理节点所拥有的线程个数为80时,就需要为服务器集群中的处理节点配置80个线程用来并行处理消息。所述消息分配规则用于表示待处理的消息由哪个线程进行处理,例如,用户A请求处理的消息由线程编号为38的线程进行处理。其中的消息分配规则在配置时需要考虑节点的综合性能指标,包括软件环境和硬件环境。如果适用于计算需求较高的场景,可以按CPU内核的数量来分配各个节点的线程个数,例如,8核的节点分配80个线程,而16核的机器分配160个线程;如果适用于内存需求较高的场景,则可以按内存参数(例如内存的性能状态和容量)来进行分配,也可以综合考虑内存、CPU、输入输出(IO)或网络等实际情况。其中的消息分配规则在配置时,则可以通过HASH取模计算来实现,例如通过如下公式获取:message_lable_id%total_thread_count==thread_id,该公式中的“message_lable_id”表示消息顺序子集的标识,同一个用户的消息被分到同一个顺序子集中,以保证同一个用户的消息可以按照顺序被处理,而不同用户的消息则不需要按照顺序来处理;其中的“total_thread_count”为每个节点的线程个数,其中的“thread_id”为线程编号。即是如果消息顺序子集的标识对总线程数取模结果为线程编号,则该消息顺序子集的消息由该线程编号对应的线程领取并处理。在具体实现时,消息分配规则可以有多种方式,可以自主选取任何能使消息分配均匀的方式实现。步骤102:依据所述线程分配规则创建每个处理节点对应的多个线程。集群中的每个处理节点在初始化时将根据线程分配规则创建若干个线程,例如每个处理节点创建50个线程等,创建的为线程后续可以根据消息分配规则从消息源中领取并处理消息。步骤103:触发所述创建的多个线程按照所述消息分配规则从消息源领取消息,并进行处理。本实施例中所提及的消息源,是在一个服务器集群中接收到的所有用户请求处理的消息的集合,任意一个线程都需要来此数据源中领取消息。其中,一个所述线程按照所述消息分配规则从消息源中领取对应的消息的步骤,参考图2所示,具体可以包括:步骤201:将请求处理消息的用户标识编号按照所述线程个数进行取模运算;在本实施例中,因为同一个用户请求的消息属于同一个顺序子集,而不同用户请求的消息属于不同的顺序子集,所以也可以按照请求处理消息的用户标识编号来对线程个数进行取模。其中,用户标识编号可以按照1、2、3等自然数的方式递增,一般可以通过数据库的自增主键来实现,可以用来标识不同的顺序子集。步骤202:线程编号与所述运算结果匹配的线程从所述消息源中领取所述用户标识触发的消息。步骤201的运算结果如果与某一个线程编号相同,则该线程编号对应的线程就负责处理上述用户标识编号对应的用户所请求的消息。可见,通过将用户的顺序自己与取模运算进行关联,可以实现消息领取的并行,即是不同的线程可以根据取模运算结果自主从消息源中领取消息。采用本发明实施例,通过建立线程与包括其需要处理消息的顺序子集的关联,从而能够实现线程领取消息过程的并行性,提高了节点处理消息的实时性,从而带来服务器集群的吞吐量和实时性的提升。参考图3,示出了本申请一种并行处理消息的方法实施例2的流程图,可以包括以下步骤:步骤301:依据服务器集群中处理节点的CPU数量和/或内存参数设置所述线程分配规则。本实施例中首先依据处理节点的CPU数量和/或内存参数等实际情况,来设置所述线程分配规则。具体设置方式可以参考实施例1中的描述,在此不再赘述。步骤302:设置消息分配规则,并将所述线程分配规则和消息分配规则存储至配置数据库。同时,在设置每个线程需要遵循的消息分配规则,并将设置好的线程分配规则和消息分配规则存储至配置数据库。步骤303:从所述配置数据库中获取线程分配规则和消息分配规则;所述线程分配规则用于表示集群中的每个处理节点所拥有的线程个数,所述消息分配规则用于表示待处理的消息由哪个线程进行处理。步骤304:依据所述线程分配规则创建每个处理节点对应的多个线程。假设线程分配规则为每个节点配置80个线程,则为服务器集群中的每个节点配置为80个线程,如果服务器集群中共10个节点,则线程编号则可以为1~800。步骤305:所述创建的多个线程按照所述消息分配规则从消息源领取消息,并进行处理。所述多个线程领取消息时,将用户标识编号按每个节点的线程个数取模,结果等于线程编号的那些消息将组成一个顺序子集,该顺序子集内的消息被该线程编号对应的线程按领取出来的顺序进行处理。例如,节点A中线程编号为5的线程,在领取时,将领取到Cust_id%800=5的用户标识编号的顺序子集中的消息,其中“Cust_id”表示用户标识编号。步骤306:对所述线程分配规则和消息分配规则进行更新。需要说明的是,在后续应用时,还可以对预先配置的线程分配规则和消息分配规则进行更新。例如,在消息量增长过快需要对服务器集群进行扩容时,可以对预置的配置数据库中的线程分配规则和消息分配规则进行修改,而这种动态更新的具体实现可以是:调用提前部署在各个节点的指令,按顺序触发集群中的各个节点处理完当前消息后不再领取消息,并按顺序重新初始化各个节点、线程分配规则及消息分配规则,在实际应用中每个节点可能停止服务一段时间(1-5分钟),但是这短暂的时间对实时性要求在分钟级(1-5分钟)的应用都不会有影响。本实施例不仅可以解决领取消息导致的影响消息处理的实时性和服务器的吞吐量的技术问题,还能够更为方便和高效的实现对服务器集群扩容,更加适应于目前网络应用场景。参考图4所示,为本申请实施例公开的并行处理消息的方法在实际应用的应用示意图,其中,一个服务器集群包括了n个处理节点,而一个处理节点则根据预置的配置数据库中的线程分配规则创建了M个线程,同时每个线程从消息源中按照消息分配规则领取消息并进行处理。对于前述的各方法实施例,为了简单描述,故将其都表述为一系列的动作组合,但是本领域技术人员应该知悉,本申请并不受所描述的动作顺序的限制,因为依据本申请,某些步骤可以采用其他顺序或者同时进行。其次,本领域技术人员也应该知悉,说明书中所描述的实施例均属于优选实施例,所涉及的动作和模块并不一定是本申请所必须的。与上述本申请一种并行处理消息的方法实施例1所提供的方法相对应,参见图5,本申请还提供了一种并行处理消息的装置实施例1,在本实施例中,该装置可以包括:获取模块501,用于从预置的配置数据库中获取线程分配规则和消息分配规则;所述线程分配规则用于表示集群中的每个处理节点所拥有的线程个数,所述消息分配规则用于表示待处理的消息由哪个线程进行处理。创建模块502,用于依据所述线程分配规则创建每个处理节点对应的多个线程。触发模块503,用于触发所述创建的多个线程按照所述消息分配规则从消息源领取消息,并进行处理。其中,参考图6所示,所述触发模块503具体可以包括运算子模块601和触发子模块602,所述运算子模块601可以用于将请求处理消息的用户标识编号按照所述线程个数进行取模运算;所述触发子模块602可以用于触发线程编号与所述运算结果匹配的线程从所述消息源中领取所述用户标识触发的消息。采用本发明实施例所述的装置,可以通过建立线程与包括其需要处理消息的顺序子集的关联,从而能够实现线程领取消息过程的并行性,提高了节点处理消息的实时性,从而带来服务器集群的吞吐量和实时性的提升。与上述本申请一种并行处理消息的方法实施例2所提供的方法相对应,参见图7,本申请还提供了一种并行处理消息的装置实施例2,在本实施例中,该装置可以包括:设置模块701,用于依据所述处理节点的CPU数量和/或内存参数设置所述线程分配规则。获取模块501,用于从预置的配置数据库中获取线程分配规则和消息分配规则;所述线程分配规则用于表示集群中的每个处理节点所拥有的线程个数,所述消息分配规则用于表示待处理的消息由哪个线程进行处理;创建模块502,用于依据所述线程分配规则创建每个处理节点对应的多个线程;触发模块503,用于触发所述创建的多个线程按照所述消息分配规则从消息源领取消息,并进行处理。更新模块702,用于对所述线程分配规则和消息分配规则进行更新。需要说明的是,当本申请所述的方法采用软件实现时,可以作为节点中新增的一个功能,也可以单独编写相应的程序,本申请不限定所述装置的实现方式。通过本实施例公开的装置,不仅可以解决领取消息导致的影响消息处理的实时性和服务器的吞吐量的技术问题,还能够更为方便和高效的实现对服务器集群扩容,更加适应于目前网络应用场景。此外,本申请实施例还公开一种并行处理消息的节点,所述节点具体可以包括:前述装置实施例1或者实施例2所述的装置。对于装置的相关介绍可以参考前述方法实施例和装置实施例,在此不再赘述。同时本申请实施例还公开一种服务器集群,所述服务器集群具体可以包括:至少两个本申请实施例公开的所述并行处理消息的节点。对于节点的相关介绍可以参考前述方法实施例和装置实施例,在此不再赘述。需要说明的是,本说明书中的各个实施例均采用递进的方式描述,每个实施例重点说明的都是与其他实施例的不同之处,各个实施例之间相同相似的部分互相参见即可。对于装置类实施例而言,由于其与方法实施例基本相似,所以描述的比较简单,相关之处参见方法实施例的部分说明即可。最后,还需要说明的是,术语“包括”、“包含”或者其任何其他变体意在涵盖非排他性的包含,从而使得包括一系列要素的过程、方法、物品或者设备不仅包括那些要素,而且还包括没有明确列出的其他要素,或者是还包括为这种过程、方法、物品或者设备所固有的要素。在没有更多限制的情况下,由语句“包括一个......”限定的要素,并不排除在包括所述要素的过程、方法、物品或者设备中还存在另外的相同要素。以上对本申请所提供的一种并行处理消息的方法、装置、节点及服务器集群进行了详细介绍,本文中应用了具体个例对本申请的原理及实施方式进行了阐述,以上实施例的说明只是用于帮助理解本申请的方法及其核心思想;同时,对于本领域的一般技术人员,依据本申请的思想,在具体实施方式及应用范围上均会有改变之处,综上所述,本说明书内容不应理解为对本申请的限制。